#include "mini_redis/AofLogger.h"
#include "mini_redis/KeyValueStore.h"
#include "mini_redis/log.h"

#include <bits/types/struct_iovec.h>
#include <cstddef>
#include <fcntl.h>
#include <mutex>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#include <errno.h>
#include <filesystem>
#include <optional>
#include <sys/stat.h>
#include <sys/uio.h>
#include <utility>
#include <vector>

namespace mini_redis {

static std::string joinPath(const std::string& dir,
                            const std::string& file)
{
    if (dir.empty())
        return file;
    if (dir.back() == '/')
        return dir + file;
    return dir + "/" + file;
}

std::string AofLogger::path() const
{
    return joinPath(opts_.dir, opts_.filename);
}

bool AofLogger::init(const AofOptions& opts, std::string& err)
{
    opts_ = opts;
    if (!opts_.enabled)
        return true;

    std::error_code ec;
    std::filesystem::create_directories(opts_.dir, ec);
    if (ec)
    {
        err = "mkdir failed: " + opts_.dir;
        return false;
    }

    fd_ = ::open(path().c_str(), O_CREAT | O_APPEND | O_WRONLY, 0644);
    if (fd_ < 0)
    {
        err = "open AOF failed: " + path();
        return false;
    }

#ifdef __linux__
    if (opts_.prealloc_bytes > 0)    // 如果配置了预分配空间大小
    {
        // 预分配文件空间（从偏移量0开始，分配指定字节数）
        // 作用：减少后续写入时的文件元数据更新开销，提高性能
        posix_fallocate(fd_, 0, static_cast<off_t>(opts_.prealloc_bytes));
    }
#endif

    running_.store(true);
    stop_.store(false);
    writer_thread_ = std::thread(&AofLogger::writerLoop, this);
    return true;
}

bool AofLogger::load(KeyValueStore& store, std::string& err)
{
    if (!opts_.enabled)
        return true;

    int rfd = ::open(path().c_str(), O_RDONLY);
    if (rfd < 0)
    {
        // 若文件不存在，不算致命错误（可能是首次启动），返回成功
        return true;
    }

    Defer d([rfd]() { ::close(rfd); });

    std::string buf;
    buf.resize(1 << 20);    // 缓冲区大小为1MB（2^20字节）
    std::string data;
    while (true)
    {
        ssize_t r = ::read(rfd, buf.data(), buf.size());
        if (r < 0)
        {    // 读取错误
            err = "read AOF failed";
            ::close(rfd);
            return false;
        }
        if (r == 0)
            break;    // 读取完毕

        // 累加内容到data
        data.append(buf.data(), static_cast<size_t>(r));
    }

    // AOF 文件中的命令通常用RESP 协议（Redis 序列化协议）存储
    size_t pos      = 0;
    auto   readLine = [&](std::string& out) -> bool {
        size_t e = data.find("\r\n", pos);
        if (e == std::string::npos)
            return false;
        out.assign(data.data() + pos, e - pos);
        pos = e + 2;
        return true;
    };

    while (pos < data.size())
    {
        // 1. 检查是否是RESP数组开头（*表示数组，对应一个命令）
        if (data[pos++] != '*')
            break;

        // 2. 读取数组长度（如SET有3个参数：SET、key、value）
        std::string line;
        if (!readLine(line))
            break;                  // 读不到长度，解析失败
        int n = std::stoi(line);    // 转换为整数（参数个数）

        // 3. 读取n个参数
        std::vector<std::string> parts;    // 存储命令的所有参数
        parts.reserve(static_cast<size_t>(n));
        for (int i = 0; i < n; ++i)
        {
            // 3.1 检查是否是批量字符串开头（$表示后面是长度）
            if (data[pos++] != '$')
            {
                err = "bad bulk";
                return false;
            }
            // 3.2 读取参数长度
            if (!readLine(line))
            {
                err = "bad bulk len";
                return false;
            }
            int len = std::stoi(line);    // 参数的字节长度

            // 3.3 检查数据是否完整（避免截断）
            if (pos + static_cast<size_t>(len) + 2 > data.size())
            {
                err = "trunc";
                return false;
            }

            // 3.4 提取参数内容（跳过最后的\r\n）
            parts.emplace_back(data.data() + pos,
                               static_cast<size_t>(len));
            pos += static_cast<size_t>(len) + 2;    // 移动指针到下一个参数
        }

        // 4. 执行解析出的命令
        if (parts.empty())
            continue;    // 空命令跳过

        std::string cmd;
        // 命令转为大写（兼容大小写，如set和SET都视为SET）
        for (char c : parts[0])
            cmd.push_back(static_cast<char>(::toupper(c)));

        // 处理SET命令（SET key value）
        if (cmd == "SET" && parts.size() == 3)
        {
            store.set(parts[1], parts[2]);    // 调用键值存储的set方法
        }
        // 处理DEL命令（DEL key1 key2 ...）
        else if (cmd == "DEL" && parts.size() >= 2)
        {
            std::vector<std::string> keys(parts.begin() + 1, parts.end());
            store.del(keys);    // 调用键值存储的del方法
        }
        // 处理EXPIRE命令（EXPIRE key seconds）
        else if (cmd == "EXPIRE" && parts.size() == 3)
        {
            int64_t sec = std::stoll(parts[2]);    // 过期时间（秒）
            store.expire(parts[1], sec);    // 调用键值存储的expire方法
        }
    }

    return true;
}

/********************************************************************/

std::string toRespArray(const std::vector<std::string>& parts)
{
    std::string out;
    out.reserve(16 * parts.size());    // 预分配内存，提升效率

    // 1. 写入数组头部：*<元素个数>\r\n
    // 例如 parts 有 3 个元素，就写入 "*3\r\n"
    out.append("*").append(std::to_string(parts.size())).append("\r\n");

    // 2. 逐个写入元素（每个元素用 RESP 批量字符串格式）
    for (const auto& p : parts)
    {
        // 批量字符串格式：$<长度>\r\n<内容>\r\n
        // 例如元素 "SET" 会写成 "$3\r\nSET\r\n"
        out.append("$").append(std::to_string(p.size())).append("\r\n");
        out.append(p).append("\r\n");
    }
    return out;
}

bool AofLogger::appendCommand(const std::vector<std::string>& parts)
{
    if (!opts_.enabled || fd_ < 0)
        return true;

    // 1、转换命令为 RESP 格式
    std::string line = toRespArray(parts);
    std::string line_copy;

    // 2、处理 AOF 重写时的增量命令
    // AOF 重写时会生成一个新的压缩文件（去除冗余命令），
    // 重写期间产生的新命令需要临时缓存（incr_cmds_），避免丢失。
    bool need_incr = rewriting_.load();    // 检查是否正在进行 AOF 重写
    if (need_incr)
        line_copy = line;    // 复制一份命令，用于后续追加到新 AOF 文件

    // 3、将命令加入写入队列
    int64_t my_seq = 0;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        pending_bytes_ += line.size();    // 更新待写入字节数
        my_seq = ++seq_gen_;    // 生成唯一序列号（用于追踪同步状态）
        queue_.push_back(AofItem{std::move(line), my_seq});    // 加入队列
    }

    // 4、缓存重写期间的增量命令
    if (need_incr)
    {
        std::lock_guard<std::mutex> lock(incr_mtx_);
        incr_cmds_.emplace_back(std::move(line_copy));    // 加入增量缓存
    }

    cv_.notify_one();    // 通知写入线程：有新命令需要处理
    // 5、处理 kAlways 模式的同步等待
    if (opts_.mode == AofMode::kAlways)
    {
        std::unique_lock<std::mutex> lock(mutex_);
        // 等待直到当前命令已同步到磁盘（last_synced_seq_ >=
        // my_seq）或需要停止
        cv_commit_.wait(lock, [&] {
            return last_synced_seq_ >= my_seq || stop_.load();
        });
    }

    return true;
}

bool AofLogger::appendRaw(const std::string& raw_resp)
{
    if (!opts_.enabled || fd_ < 0)
        return true;
    std::string line_copy;
    bool        need_incr = rewriting_.load();
    if (need_incr)
        line_copy = raw_resp;    // 复制用于增量缓冲

    int64_t my_seq = 0;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        pending_bytes_ += raw_resp.size();
        my_seq = ++seq_gen_;
        queue_.push_back(AofItem{std::string(raw_resp), my_seq});
    }

    if (need_incr)
    {
        std::lock_guard<std::mutex> lock(incr_mtx_);
        incr_cmds_.emplace_back(std::move(line_copy));
    }

    cv_.notify_one();
    if (opts_.mode == AofMode::kAlways)
    {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_commit_.wait(lock, [&] {
            // 等待直到当前命令已同步到磁盘（last_synced_seq_ >=
            // my_seq）或需要停止
            return last_synced_seq_ >= my_seq || stop_.load();
        });
    }

    return true;
}

}    // namespace mini_redis

namespace mini_redis {

static std::string joinPath(const std::string& dir,
                            const std::string& file);

// writeAllFD 是一个封装了系统 write
// 调用的工具函数，核心目的是确保要写入的数据被 “完整写入”
// 到文件描述符（避免系统 write 调用 “部分写入” 导致数据丢失），
// 是 AOF 持久化中保证数据可靠性的关键底层函数。
static bool writeAllFD(int fd, const char* data, size_t len)
{
    size_t off = 0;
    while (off < len)
    {
        ssize_t w = ::write(fd, data + off, len - off);
        if (w > 0)
        {
            off += static_cast<size_t>(w);
            continue;
        }
        if (w < 0 && (errno == EINTR || errno == EAGAIN))
        {
            continue;
        }
        return false;
    }
    return true;
}

#if 0
    writerLoop 是 AOF 写入线程（writer_thread_）的核心函数，
负责将主线程放入队列的命令异步写入 AOF 文件，
并根据配置的同步模式（kAlways/kEverySec）保证数据可靠性。
整个函数围绕 “高效批量写入” 和 “按需同步磁盘” 设计，
避免阻塞主线程的同时确保数据安全。

#endif
void AofLogger::writerLoop()
{
    // 1. 初始化批量写入参数
    // 批量写入的目标字节数（默认64KB）
    const size_t kBatchBytes =
        opts_.batch_bytes > 0 ? opts_.batch_bytes : (64 * 1024);

    // 一次最多处理的IO向量数量（限制单次写入的命令数）
    const int kMaxIov = 64;

    // 等待新命令的超时时间（默认1毫秒）
    const auto kWaitNs = std::chrono::microseconds(
        opts_.batch_wait_us > 0 ? opts_.batch_wait_us : 1000);

    // 本地缓存，临时存储要写入的命令（减少锁持有时间）
    std::vector<AofItem> local;
    local.reserve(256);    // 预分配空间，提升效率

    while (!stop_.load())
    {
        // 若收到暂停信号（如AOF重写切换文件时）
        if (pause_writer_.load())
        {
            std::unique_lock<std::mutex> lock(pause_mtx_);
            writer_is_paused_ = true;

            // 通知等待者（如重写线程）：我已暂停
            cv_pause_.notify_all();

            // 等待暂停信号解除（pause_writer_=false）或需要退出
            cv_pause_.wait(lock, [&]() {
                return !pause_writer_.load() || stop_.load();
            });

            writer_is_paused_ = false;

            if (stop_.load())
                break;
        }

        // 从命令队列批量取命令
        local.clear();
        size_t bytes = 0;
        {
            std::unique_lock<std::mutex> lock(mutex_);

            // 若队列为空，等待新命令（最多等kWaitNs时间）
            if (queue_.empty())
            {
                cv_.wait_for(lock, kWaitNs, [&]() {
                    return stop_.load() || !queue_.empty();
                });
            }

            // 从队列取命令，直到满足以下任一条件：
            // 1. 队列空了；
            // 2. 总字节数达到kBatchBytes；
            // 3.命令数达到kMaxIov
            while (!queue_.empty() && (bytes < kBatchBytes) &&
                   (int)local.size() < kMaxIov)
            {
                local.emplace_back(std::move(queue_.front()));
                bytes += local.back().data.size();
                queue_.pop_front();
            }

            // 更新待写入字节数（pending_bytes_）
            if (pending_bytes_ >= bytes)
                pending_bytes_ -= bytes;
            else
                pending_bytes_ = 0;
        }

        // 没取到命令
        if (local.empty())
        {
            // 若为kEverySec模式，检查是否需要周期性同步磁盘
            if (opts_.mode == AofMode::kEverySec)
            {
                auto now = std::chrono::steady_clock::now();    // 当前时间

                auto interval = std::chrono::milliseconds(
                    opts_.sync_interval_ms > 0
                        ? opts_.sync_interval_ms
                        : 1000);    // 同步间隔（默认1秒）

                // 超过间隔，触发同步
                if (now - last_sync_tp_ >= interval)
                {
                    if (fd_ >= 0)
                    {
                        ::fdatasync(fd_);    // 同步到磁盘
                    }
                    last_sync_tp_ = now;    // 更新最后同步时间
                }
            }
            continue;    // 回到循环开头，继续等命令
        }

        // 组装 IO 向量，准备批量写入
        struct iovec iov[kMaxIov];

        int iovcnt = 0;
        for (auto& it : local)
        {
            if (iovcnt >= kMaxIov)
                break;

            iov[iovcnt].iov_base = const_cast<char*>(it.data.data());
            iov[iovcnt].iov_len  = it.data.size();
            iovcnt++;
        }

        // 批量写入文件
        int    start_idx = 0;
        size_t start_off = 0;
        while (start_idx < iovcnt)
        {
            ssize_t w = ::writev(fd_, &iov[start_idx], iovcnt - start_idx);
            if (w < 0)    // 写入出错
            {
                ::usleep(1000);    // 休眠1毫秒再重试（避免忙等）
                break;
            }

            size_t rem = static_cast<size_t>(w);

            // 处理“部分写入”：比如要写1000字节，只写了300字节
            while (rem > 0 && start_idx < iovcnt)
            {
                // 当前iovec剩余可写长度
                size_t avail = iov[start_idx].iov_len - start_off;

                // 本次写入的字节数小于当前iovec剩余长度
                if (rem < avail)
                {
                    start_off += rem;    // 偏移量后移

                    iov[start_idx].iov_base =
                        static_cast<char*>(iov[start_idx].iov_base) + rem;

                    // 剩余长度更新
                    iov[start_idx].iov_len = avail - rem;

                    rem = 0;    // 本次写入处理完
                }
                else    // 本次写入的字节数覆盖当前iovec剩余部分
                {
                    rem -= avail;     // 剩余待处理字节数减少
                    ++start_idx;      // 处理下一个iovec
                    start_off = 0;    // 偏移量重置
                }

                if (w == 0)
                    break;
            }
        }

#ifdef __linux__
        if (opts_.use_sync_file_range && bytes >= opts_.sfr_min_bytes)
        {
            off_t cur = ::lseek(fd_, 0, SEEK_END);
            if (cur > 0)
            {
                off_t start = cur - static_cast<off_t>(bytes);
                if (start < 0)
                    start = 0;

                // 通知内核异步写回[start, start+bytes)区间的数据
                (void)::sync_file_range(fd_, start,
                                        static_cast<off_t>(bytes),
                                        SYNC_FILE_RANGE_WRITE);
            }
        }
#endif

        // 模式处理
        if (opts_.mode == AofMode::kAlways)
        {
            ::fdatasync(fd_);
#ifdef __linux__
            // 通知内核释放已同步的内存页（减少内存占用）
            if (opts_.fadvise_dontneed_after_sync)
            {
                off_t cur2 = ::lseek(fd_, 0, SEEK_END);
                if (cur2 > 0)
                {
                    (void)::posix_fadvise(fd_, 0, cur2,
                                          POSIX_FADV_DONTNEED);
                }
            }
#endif
            // 更新“最后同步的序列号”，并唤醒等待的主线程
            int64_t max_seq = 0;

            // 取这批命令的最大序列号
            for (auto& it : local)
            {
                max_seq = std::max(max_seq, it.seq);
            }

            {
                std::lock_guard<std::mutex> lg(mutex_);
                last_synced_seq_ =
                    std::max(last_synced_seq_, max_seq);    // 更新同步进度
            }

            // 唤醒主线程（主线程在kAlways模式下会等待同步完成）
            cv_commit_.notify_all();
        }
        else if (opts_.mode == AofMode::kEverySec)
        {
            auto now      = std::chrono::steady_clock::now();
            auto interval = std::chrono::milliseconds(
                opts_.sync_interval_ms > 0 ? opts_.sync_interval_ms
                                           : 1000);

            if (now - last_sync_tp_ >= interval)
            {
                ::fdatasync(fd_);
                last_sync_tp_ = now;
#ifdef __linux__
                if (opts_.fadvise_dontneed_after_sync)
                {
                    off_t cur3 = ::lseek(fd_, 0, SEEK_END);
                    if (cur3 > 0)
                    {
                        (void)::posix_fadvise(fd_, 0, cur3,
                                              POSIX_FADV_DONTNEED);
                    }
                }
#endif
            }
        }
    }

    // 退出前flush
    if (fd_ >= 0)
    {
        while (true)
        {
            std::vector<AofItem> rest;
            size_t               bytes = 0;
            {
                std::lock_guard<std::mutex> lock(mutex_);
                while (!queue_.empty() && (int)rest.size() < 64)
                {
                    rest.emplace_back(std::move((queue_.front())));
                    bytes += rest.back().data.size();
                    queue_.pop_front();
                }
                if (pending_bytes_ >= bytes)
                    pending_bytes_ -= bytes;
                else
                    pending_bytes_ = 0;
            }

            if (rest.empty())
                break;

            struct iovec iov2[64];
            int          n = 0;
            for (auto& it : rest)
            {
                iov2[n].iov_base = const_cast<char*>(it.data.data());
                iov2[n].iov_len  = it.data.size();
                ++n;
            }

            int    start_idx2 = 0;
            size_t start_off2 = 0;

            while (start_idx2 < n)
            {
                ssize_t w2 =
                    ::writev(fd_, &iov2[start_idx2], n - start_idx2);
                if (w2 < 0)
                {
                    if (errno == EINTR || errno == EAGAIN)
                        continue;
                    else
                        break;
                }

                size_t rem2 = static_cast<size_t>(w2);
                while (rem2 > 0 && start_idx2 < n)
                {
                    size_t avail2 = iov2[start_idx2].iov_len - start_off2;
                    if (rem2 < avail2)
                    {
                        start_off2 += rem2;
                        iov2[start_idx2].iov_base =
                            static_cast<char*>(iov2[start_idx2].iov_base) +
                            rem2;

                        iov2[start_idx2].iov_len = avail2 - rem2;

                        rem2 = 0;
                    }
                    else
                    {
                        rem2 -= avail2;
                        ++start_idx2;
                        start_off2 = 0;
                    }
                }

                if (w2 == 0)
                    break;
            }
        }
    }

    ::fdatasync(fd_);
}

bool AofLogger::bgRewrite(KeyValueStore& store, std::string& err)
{
    if (!opts_.enabled)
    {
        err = "aof disabled";
        return false;
    }

    // 检查是否已有重写在运行
    bool expected = false;
    if (!rewriting_.compare_exchange_strong(expected, true))
    {
        err = "rewrite already running";
        return false;
    }

    rewriter_thread_ = std::thread(&AofLogger::rewriterLoop, this, &store);
    return true;
}

void AofLogger::rewriterLoop(KeyValueStore* store)
{
    // 生成临时文件路径，避免重写出错损坏旧文件
    std::string tmp_path =
        joinPath(opts_.dir, opts_.filename + ".rewrite.tmp");
    int wfd = ::open(tmp_path.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0644);
    if (wfd < 0)
    {
        rewriting_.store(false);
        return;
    }

    // 遍历内存快照，生成精简命令集（重写核心）
    {
        // 获取内存中所有String类型数据的快照（键值对+过期时间）
        auto snap = store->snapshot();
        for (const auto& kv : snap)    // 遍历每个键值对
        {
            const std::string& k = kv.first;    // 键

            // 值+过期信息（r.value是值，r.expire_at_ms是过期时间戳）
            const auto& r = kv.second;

            // 生成精简命令：直接用SET写入最终值（替代所有历史修改命令）
            std::vector<std::string> parts = {"SET", k, r.value};
            std::string line = toRespArray(parts);    // 转换为RESP协议格式

            writeAllFD(wfd, line.data(),
                       line.size());    // 写入临时文件（确保完整写入）

            // 处理过期时间：若该键有过期时间，生成EXPIRE命令
            if (r.expire_at_ms > 0)
            {
                // 计算当前剩余TTL（过期时间戳 - 当前时间戳，转换为秒）
                int64_t now =
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::steady_clock::now()
                            .time_since_epoch())
                        .count();

                int64_t ttl = (r.expire_at_ms - now) / 1000;
                if (ttl < 1)
                    ttl = 1;    // 避免TTL为0或负数（确保过期生效）

                // 生成EXPIRE命令并写入临时文件
                std::vector<std::string> e = {"EXPIRE", k,
                                              std::to_string(ttl)};

                std::string el = toRespArray(e);
                writeAllFD(wfd, el.data(), el.size());
            }
        }
    }

    // Hash
    {
        auto snap = store->snapshotHash();
        for (const auto& kv : snap)
        {
            const std::string& key = kv.first;
            const auto&        h   = kv.second;
            for (const auto& fv : h.fields)
            {
                std::vector<std::string> parts = {"HSET", key, fv.first,
                                                  fv.second};

                std::string line = toRespArray(parts);
                writeAllFD(wfd, line.data(), line.size());
            }
            if (h.expire_at_ms > 0)
            {
                int64_t now =
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::steady_clock::now()
                            .time_since_epoch())
                        .count();

                int64_t ttl = (h.expire_at_ms - now) / 1000;
                if (ttl < 1)
                    ttl = 1;
                std::vector<std::string> e = {"EXPIRE", key,
                                              std::to_string(ttl)};

                std::string el = toRespArray(e);
                writeAllFD(wfd, el.data(), el.size());
            }
        }
    }

    // ZSet
    {
        auto snap = store->snapshotZSet();
        for (const auto& flat : snap)
        {
            for (const auto& it : flat.items)
            {
                std::vector<std::string> parts = {
                    "ZADD", flat.key, std::to_string(it.first), it.second};
                std::string line = toRespArray(parts);
                writeAllFD(wfd, line.data(), line.size());
            }
            if (flat.expire_at_ms > 0)
            {
                int64_t now =
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::steady_clock::now()
                            .time_since_epoch())
                        .count();
                int64_t ttl = (flat.expire_at_ms - now) / 1000;
                if (ttl < 1)
                    ttl = 1;
                std::vector<std::string> e  = {"EXPIRE", flat.key,
                                              std::to_string(ttl)};
                std::string              el = toRespArray(e);
                writeAllFD(wfd, el.data(), el.size());
            }
        }
    }

    // 进入切换阶段：暂停 writer，写入最终增量并原子替换
    pause_writer_.store(true);
    {
        std::unique_lock<std::mutex> lk(pause_mtx_);
        // 如果不等待写入线程暂停，可能出现
        // “新命令同时写入旧文件和增量列表”
        // 的情况，导致数据重复或丢失。必须确保写入线程完全停止写入旧文件后，再处理增量命令。
        cv_pause_.wait(lk, [&] { return writer_is_paused_; });
    }

    // 在 writer 暂停期间，阻塞增量缓冲追加，确保没有遗漏
    {
        std::lock_guard<std::mutex> lg(
            incr_mtx_);    // 保护增量命令列表（避免主线程同时写入）
        for (const auto& s : incr_cmds_)    // 遍历所有重写期间的新命令
        {
            writeAllFD(wfd, s.data(), s.size());    // 追加到临时文件
        }
        incr_cmds_.clear();    // 清空增量列表，避免后续重复写入
    }
    ::fdatasync(wfd);    // 强制将临时文件数据刷到磁盘（确保不丢失）

    // 原子替换并切换 fd
    {
        std::string final_path = path();
        ::close(fd_);
        ::close(wfd);
        ::rename(tmp_path.c_str(), final_path.c_str());
        fd_ = ::open(final_path.c_str(), O_CREAT | O_APPEND | O_WRONLY,
                     0644);
        // fsync 目录，保证 rename 持久
        int dfd = ::open(opts_.dir.c_str(), O_RDONLY);
        if (dfd >= 0)
        {
            ::fsync(dfd);
            ::close(dfd);
        }
    }

    pause_writer_.store(false);
    cv_pause_.notify_all();
    // 清理增量
    {
        std::lock_guard<std::mutex> lg(incr_mtx_);
        incr_cmds_.clear();
    }
    rewriting_.store(false);
}

void AofLogger::shutdown()
{
    if (!running_.load())
        return;    // 避免重复调用
    running_.store(false);
    stop_.store(true);
    cv_.notify_all();    // 唤醒写入线程
    if (writer_thread_.joinable())
    {
        writer_thread_.join();
    }
    if (rewriter_thread_.joinable())
    {    // 补充重写线程的 join
        rewriter_thread_.join();
    }
    if (fd_ >= 0)
    {
        ::fdatasync(fd_);
        ::close(fd_);
        fd_ = -1;
    }
}

}    // namespace mini_redis